package com.dtwave.lnstreaming.filter.function;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.utils.JsonUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;

public class FlatMapFunction extends RichFlatMapFunction<String, JSONObject> {
    @Override
    public void flatMap(String jsonString, Collector<JSONObject> collector) throws Exception {
        JSONObject jsonObject = JsonUtils.getJSONObject(jsonString);
        if (jsonObject != null) {
            collector.collect(jsonObject);
        }
    }
}
